package goldberg.drew.twitter;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URL;

import java.util.concurrent.Callable;
import java.util.concurrent.Future;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class ConcurrentTwitterV2 extends AbstractTwitter{
	
	private static int poolSize;
	private static ExecutorService executorPool;	
	private static final ReadWriteLock monitor = new ReentrantReadWriteLock();
	final List<Callable<Integer>> partitions = new ArrayList<Callable<Integer>>();
	private static int counter = 1;
	/**
	* Empty Constructor
	*/
	private ConcurrentTwitterV2() {
	}
	
	/**
	* Creates the directory hierarchy and textfile for each user	
	* usersConcurrent/userName/userName.txt
	* intializes the threadPool executorPool
	*/
	private void init() {

		ConcurrentTwitterV2.createUsers();
		poolSize = calculateThreadPoolSize();
		executorPool = Executors.newFixedThreadPool(poolSize);
		runThreads();	
	}
	
	public static ConcurrentTwitterV2 create() {
		final ConcurrentTwitterV2 concurrentTwitterV2 = new ConcurrentTwitterV2();
			concurrentTwitterV2.init();
		return concurrentTwitterV2;
	}

	public int calculateThreadPoolSize() {
		final int numberOfCores = Runtime.getRuntime().availableProcessors();
		final double blockingCoefficient = 0.9;
		final int threadPoolSize = (int) (numberOfCores/(1 - blockingCoefficient));
		System.out.println("The number of cores is " + numberOfCores);
		System.out.println("The blocking coefficient is " + blockingCoefficient);
		System.out.println("The number of threads is " + threadPoolSize);		
		return threadPoolSize;	
	}	
	

	public void runThreads() {
		for(final String key : users.keySet()) {										
			getTweets(key, users.get(key));
			System.out.println("The key value is " + key);
		}
					
	}
               
	public void getTweets(final String userName, final File file) {
	    
    		for(int i = 0; i < 10; i++) {
			final int pageNumber = i+1;
			executorPool.submit(new Runnable (){
				public void run() {										
					monitor.writeLock().lock();
					monitor.readLock().lock();					
					try {	
						System.out.println("Retreiving User " + userName + " page " + pageNumber);				
						final BufferedInputStream in = new BufferedInputStream(new URL("http://api.twitter.com/1/statuses/user_timeline.xml?id="+userName+"&count=50&page="+pageNumber).openStream());
						final FileOutputStream fos = new FileOutputStream(file.toString(), true);
						final BufferedOutputStream bout = new BufferedOutputStream(fos, 1024);
						byte[] data = new byte[1024];
						int x = 0;		
						System.out.println("Writing to user " + userName + ". Write count " + pageNumber);				
						while((x=in.read(data,0,1024))>0){
						   bout.write(data, 0, x);
						}
						bout.close();
						in.close();					
					}
					catch (IOException e) {
						System.out.println("Failed at userName " + userName + " page " + counter);
						e.printStackTrace();
					}
					finally{
						monitor.readLock().unlock();
						monitor.writeLock().unlock();					
					}
				}
			
			});
		}
		
	 }

	public static void main(String[] args) {
	
		ConcurrentTwitterV2 concurrentTwitter = new ConcurrentTwitterV2();
		final long startTime = System.nanoTime();		
		concurrentTwitter.create();
		executorPool.shutdown();
		try{
			executorPool.awaitTermination(1000,TimeUnit.SECONDS);		
	    	}
		catch(InterruptedException e){}
		final long finishTime = System.nanoTime();
		if(concurrentTwitter.executorPool.isTerminated())
			System.out.println("If all the tasks have been completed properly.\nTime take in seconds " + (finishTime - startTime / 1.0e9));
		else{
			System.out.println("This was printed because the executor pool threads haven't completed yet.\nTime take in seconds " + (finishTime - startTime / 1.0e9));
		}
        }
}
